package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"io/ioutil"
	"net/http"
	url2 "net/url"
	"sync"
	"time"
)

type CreateTopicRequest struct {
	Name  string `json:"name"`
	Uid   string `json:"uid"`
	Topic string `json:"topic"`
	Type  int    `json:"type"`
}

type GetTopicRes struct {
	Name       string `json:"name"`
	Msg        string `json:"msg"`
	Online     bool   `json:"online"`
	OnlineNum  int    `json:"onlineNum"`
	PubOnline  bool   `json:"pubOnline"`
	DeviceType string `json:"deviceType"`
	Share      bool   `json:"share"`
	Group      string `json:"group"`
	Room       string `json:"room"`
	Time       string `json:"time"`
	Unix       int    `json:"unix"`
	CreateTime string `json:"createTime"`
}

type bafaresp struct {
	Code int             `json:"code"`
	Msg  string          `json:"msg"`
	Data json.RawMessage `json:"data"`
}

func NewBafaClient(key string) (*BafaClient, error) {
	opts := mqtt.NewClientOptions()
	opts.AddBroker("tcp://bemfa.com:9501")
	opts.SetClientID(key)
	opts.SetUsername("def")
	opts.SetPassword("def")
	opts.SetAutoReconnect(true)
	opts.SetMaxReconnectInterval(30 * time.Second)
	opts.SetConnectTimeout(10 * time.Second)
	opts.SetWriteTimeout(5 * time.Second)
	opts.SetKeepAlive(60 * time.Second)
	opts.SetPingTimeout(10 * time.Second)
	client := &BafaClient{
		tk:   key,
		subs: make(map[string]mqtt.MessageHandler),
		opts: opts,
	}
	// 创建并连接客户端
	opts.SetConnectionLostHandler(client.connectionLostHandler)
	opts.SetOnConnectHandler(client.onConnectHandler)
	client.client = mqtt.NewClient(opts)
	if token := client.client.Connect(); token.Wait() && token.Error() != nil {
		return nil, token.Error()
	}
	logger.Printf("已连接到巴法云MQTT服务器")
	_ = client.ClearTopic()
	return client, nil
}

type BafaClient struct {
	client     mqtt.Client
	tk         string
	subs       map[string]mqtt.MessageHandler
	opts       *mqtt.ClientOptions
	subsMutex  sync.RWMutex
	connecting bool
	connMutex  sync.Mutex
}

func (c *BafaClient) Sub(topic string, cb mqtt.MessageHandler) error {
	c.subsMutex.Lock()
	c.subs[topic] = cb
	c.subsMutex.Unlock()

	if !c.client.IsConnected() {
		return fmt.Errorf("not connected")
	}
	if token := c.client.Subscribe(topic, 0, cb); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

func (c *BafaClient) connectionLostHandler(client mqtt.Client, err error) {
	logger.Printf("Connection lost: %v", err)
}

func (c *BafaClient) onConnectHandler(client mqtt.Client) {
	logger.Println("Reconnected to MQTT server")
	c.resubscribe()
}

func (c *BafaClient) resubscribe() {
	c.subsMutex.RLock()
	defer c.subsMutex.RUnlock()

	for topic, handler := range c.subs {
		if token := c.client.Subscribe(topic, 0, handler); token.Wait() && token.Error() != nil {
			logger.Printf("Failed to resubscribe to %s: %v", topic, token.Error())
		}
	}
}

func (c *BafaClient) CreateTopic(name, topic string) error {
	uri := "https://pro.bemfa.com/v1/createTopic"
	query := CreateTopicRequest{
		Name:  name,
		Uid:   c.tk,
		Topic: topic,
		Type:  1,
	}
	marshal, err := json.Marshal(query)
	if err != nil {
		return err
	}
	req, err := http.NewRequest(http.MethodPost, uri, bytes.NewReader(marshal))
	if err != nil {
		return err
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return fmt.Errorf("请求发送失败: %w", err)
	}
	defer resp.Body.Close()
	// 读取响应
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("读取响应失败: %w", err)
	}
	// 解析响应
	var apiResp bafaresp
	if err := json.Unmarshal(body, &apiResp); err != nil {
		return fmt.Errorf("解析响应失败: %w", err)
	}
	if apiResp.Code != 0 {
		return fmt.Errorf("API错误: %s", apiResp.Msg)
	}
	return nil
}

type ListTopicRow struct {
	Data []struct {
		Topic      string `json:"topic"`
		Type       int    `json:"type"`
		Share      bool   `json:"share"`
		Time       string `json:"time"`
		Unix       int    `json:"unix"`
		Online     bool   `json:"online"`
		Name       string `json:"name"`
		Room       string `json:"room"`
		Msg        string `json:"msg"`
		PubOnline  bool   `json:"pubOnline"`
		DeviceType string `json:"deviceType"`
		Group      string `json:"group"`
		CreateTime string `json:"createTime"`
	} `json:"data"`
}

func (c *BafaClient) ClearTopic() error {
	var query = make(url2.Values)
	query.Add("openID", c.tk)
	query.Add("type", fmt.Sprint(1))
	url := fmt.Sprintf("%s/allTopic?%s", "https://apis.bemfa.com/vb/api/v2", query.Encode())
	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return err
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return fmt.Errorf("请求发送失败: %w", err)
	}
	defer resp.Body.Close()
	// 读取响应
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("读取响应失败: %w", err)
	}
	// 解析响应
	var apiResp bafaresp
	var resps ListTopicRow
	if err := json.Unmarshal(body, &apiResp); err != nil {
		return fmt.Errorf("解析响应失败: %w", err)
	}
	if apiResp.Code != 0 {
		return fmt.Errorf("API错误: %s", apiResp.Msg)
	}
	err = json.Unmarshal(apiResp.Data, &resps)
	if err != nil {
		return err
	}
	for _, v := range resps.Data {
		if !v.Online {
			err = c.DeleteTopic(v.Topic, v.Type)
			if err == nil {
				logger.Printf("清理未在线的topic %s", v.Name)
			} else {
				logger.Printf("删除topic %s 失败: %v", v.Name, err)
			}
		}
	}
	return nil
}

type DeleteTopicRequest struct {
	Uid   string `json:"uid"`
	Topic string `json:"topic"`
	Type  int    `json:"type"`
}

func (c *BafaClient) DeleteTopic(topic string, deviceType int) error {
	uri := "https://pro.bemfa.com/v1/deleteTopic"
	query := DeleteTopicRequest{
		Uid:   c.tk,
		Topic: topic,
		Type:  deviceType,
	}
	marshal, err := json.Marshal(query)
	if err != nil {
		return err
	}
	req, err := http.NewRequest(http.MethodPost, uri, bytes.NewReader(marshal))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json; charset=utf-8")

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return fmt.Errorf("请求发送失败: %w", err)
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("读取响应失败: %w", err)
	}

	var apiResp bafaresp
	if err := json.Unmarshal(body, &apiResp); err != nil {
		return fmt.Errorf("解析响应失败: %w", err)
	}
	if apiResp.Code != 0 {
		return fmt.Errorf("API错误: %s", apiResp.Msg)
	}
	return nil
}

func (c *BafaClient) GetTopic(topic string) (*GetTopicRes, error) {
	var query = make(url2.Values)
	query.Add("openID", c.tk)
	query.Add("type", fmt.Sprint(1))
	query.Add("topic", topic)
	url := fmt.Sprintf("%s/topicInfo?%s", "https://apis.bemfa.com/vb/api/v2", query.Encode())
	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("请求发送失败: %w", err)
	}
	defer resp.Body.Close()
	// 读取响应
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("读取响应失败: %w", err)
	}
	// 解析响应
	var apiResp bafaresp
	var response GetTopicRes
	if err := json.Unmarshal(body, &apiResp); err != nil {
		return nil, fmt.Errorf("解析响应失败: %w", err)
	}
	if apiResp.Code != 0 {
		return nil, fmt.Errorf("API错误: %s", apiResp.Msg)
	}
	err = json.Unmarshal(apiResp.Data, &response)
	if err != nil {
		return nil, err
	}
	return &response, nil
}
